package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.utils.DimUtil;
import com.atguigu.gmall.realtime.utils.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.util.Collections;
import java.util.concurrent.ExecutorService;

/**
 * Author: Felix
 * Date: 2022/3/20
 * Desc: 维度数据的异步关联
 * 模板方法设计模式
 *      在父类中定义完成某一个功能的核心算法的骨架（步骤），但是具体的实现延迟到子类中去完成
 *      在不改变父类核心算法骨架的前提下，每一个子类都可以有自己不同的实现
 *
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T,T> implements DimJoinFunction<T>{
    private ExecutorService executorService;
    private String tableName;

    public DimAsyncFunction(String tableName){
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        executorService = ThreadPoolUtil.getInstance();
    }

    @Override
    public void asyncInvoke(T obj, ResultFuture<T> resultFuture) throws Exception {
        //开启多个线程
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                //发送异步请求,完成维度关联
                try {
                    long start = System.currentTimeMillis();
                    //1.根据流中处理的元素obj，获取要关联的维度的主键
                    String key = getKey(obj);
                    //2.根据维度主键获取维度对象
                    JSONObject dimInfoJsonObj = DimUtil.getDimInfo(tableName, key);
                    //3.根据维度获取获取相关的维度属性  补充到流中处理的元素obj上
                    if(dimInfoJsonObj != null && dimInfoJsonObj.size() > 0){
                        join(dimInfoJsonObj,obj);
                    }
                    long end = System.currentTimeMillis();
                    //获取数据库交互的结果并发送给 ResultFuture的回调函数
                    resultFuture.complete(Collections.singleton(obj));

                    System.out.println("异步维度关联耗时:" +(end - start) + "毫秒");
                } catch (Exception e) {
                    e.printStackTrace();
                    throw new RuntimeException("~~~异步关联维度数据失败~~~");
                }
            }
        });

    }
}
